package flow;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.function.Function;

/**
 * @author： zhangnan
 * @date： 2025/5/21 16:28
 * @description： Reactive-Stream
 * @modifiedBy：
 * @version: 1.0
 */
@Slf4j
public class FlowDemo {

    /**
     * 之前的叫命令式编程
     * 9以后 引入了 响应是编程(或 声明式编程)  在 java.util.concurrent.Flow 中
     * <p>
     * 响应式编程
     * 1、底层:基于数据缓冲队列 + 消息驱动模型 + 异步回调机制
     * 2、编码:流式编程 + 链式调用 +声明式API
     * 3、效果:优雅全异步 + 消息实时处理 + 高吞吐量 + 占用少量资源
     * <p>
     * <p>
     * 响应式编程目的：通过全异步的方式、加缓存区构建一个实时的数据流系统
     * Reactive Stream 提供了本地化大型分布式的响应式系统的解决方案
     * <p>
     * <p>
     * Reactive Streams 是JVM面向流的库的标准和规范
     * JVN底层对于整个发布订阅关系做好了 异步 + 缓存区 处理 = 响应式系统
     * 1. 处理可能无限数量的元素
     * 2. 有序
     * 3. 在组件之间异步传递元素
     * 4. 强制性 非阻塞 背压模式
     * 1. 正压：生产者的数据注解给到消费者，压力在消费者一端
     * 2. 背压：生产者的数据放入缓冲区中缓存起来，消费者根据自己的能力逐个获取并处理这些请求
     * <p>
     * 数据是自流动的，而不是靠迭代被动流动，
     * 推拉模型:
     * 推:流模式;上游有数据，自动推给下游，
     * 拉:迭代器;自己遍历，自己拉取;
     * <p>
     * Reactive Streams 核心接口，坐标：java.util.concurrent.Flow
     * Publisher：发布者 - 产生数据流
     * Subscriber：订阅者 - 消费数据流
     * Subscription：订阅关系 - 发布者 和 订阅者 之间的绑定关系
     * 订阅关系是发布者和订阅者之间的关键接口。订阅者通过订阅来表示对发布者产生的数据的兴趣。订阅者可以请求一定数量的元素，也可以取消订阅。
     * Processor：处理器
     * 处理器是同时实现了发布者和订阅者接口的组件。
     * 它可以接收来自一个发布者的数据，进行处理，并将结果发布给下一个订阅者。
     * 处理器在Reactor中充当中间环节，代表一个处理阶段，允许你在数据流中进行转换、过滤和其他操作。
     * <p>
     * 图例
     * +---+
     * |   |  +-----------+
     * |   |  | publisher |    数据发布到 dataBuffer
     * |   |  |  发布者    |
     * |   |  +-----------+
     * |   |       |
     * |   |       |
     * |   |  +-----------+     {
     * |数 |  | Processor |        处理器既是
     * |据 |  |   处理器   |              订阅者-subscriber,
     * |管 |  +-----------+              又是 发布者-publisher
     * |道 |       |
     * |   |       |
     * |   |  +-----------+
     * |   |  | Processor |
     * |   |  |  处理器    |
     * |   |  +-----------+      }
     * |   |       |
     * |   |       |
     * |   |  +------------+
     * |   |  | Subscriber |
     * |   |  |   订阅者    |
     * |   |  +------------+
     * +---+
     */


    // 自定义转换处理器
    static class TransformProcessor<T, R> extends SubmissionPublisher<R> implements Flow.Processor<T, R> {

        private final Function<T,R> transformFunction; // 转换函数
        private Flow.Subscription subscription;  // 上游的订阅对象，绑定的订阅者

        public TransformProcessor (Function<T,R> transformFunction){
            this.transformFunction = transformFunction;
        }

        // 订阅发生时，执行该方法
        @Override
        public void onSubscribe(Flow.Subscription subscription) {
//            log.debug("订阅开始了：{}" + subscription);
            this.subscription = subscription;
            subscription.request(1); // 请求第一个数据项，被咬
        }

        // 接收到数据时调用
        @Override
        public void onNext(T item) {
//            log.debug("接收数据：{}",item);
            // 应用转换函数处理数据
            R transformedItem = transformFunction.apply(item);
            submit(transformedItem); // 将转换后的数据提交给下游
            subscription.request(1); // 请求下一个数据项（背压控制）
        }

        // 发生错误时调用
        @Override
        public void onError(Throwable throwable) {
//            log.debug("出现异常");
            throwable.printStackTrace(); // 打印错误堆栈
            closeExceptionally(throwable); // 异常关闭发布者
        }

        // 数据处理完成时调用
        @Override
        public void onComplete() {
//            log.debug("数据处理结束");
            close(); // 正常关闭发布者
        }
    }

    /**
     * 自定义简单订阅者
     */
    static class SimpleSubscriber implements Flow.Subscriber<String> {

        private Flow.Subscription subscription; // 订阅对象关系

        // 当订阅建立时调用
        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            this.subscription = subscription;
//            log.debug("订阅已建立");
            subscription.request(1); // 初始请求一个数据
        }

        // 接收到数据时调用
        @Override
        public void onNext(String item) {
//            log.debug("订阅者接收到数据：{}" ,item);
            subscription.request(1);  // 再请求1个数据（背压控制）
        }

        // 发生错误时调用
        @Override
        public void onError(Throwable throwable) {
            throwable.printStackTrace();
        }


        // 数据处理完成时
        @Override
        public void onComplete() {
//            log.debug("订阅者数据处理完成");
        }
    }

    public static void main(String[] args) throws InterruptedException, IOException {

        // 1. 定义一个发布者，发布数据
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();
        // 2. 定义几个处理器模拟对数据进行操作
        TransformProcessor<Integer,Integer> multiplier = new TransformProcessor<>(n -> n * 2);
        TransformProcessor<Integer,String> stringConverter = new TransformProcessor<>(n -> "Processed-" + n);
        // 3. 定义一个消费者
        SimpleSubscriber subscriber = new SimpleSubscriber();
        // 4. 构建处理链 发布者 -> 乘法处理器 -> 字符串转换处理器 -> 订阅者
        publisher.subscribe(multiplier);
        multiplier.subscribe(stringConverter);
        stringConverter.subscribe(subscriber);
        // 5. 发布者发布数据
//        log.debug("开始发布数据...");
        for (int i = 1; i <= 5; i++) {
//            log.debug("发布原始数据:{}",i);
            publisher.submit(i);
            Thread.sleep(2000);
        }

        // 关闭发布者
        publisher.close();

        // 6. 阻塞主线程
        System.in.read();

    }

}
